package it.uni.bo.monitoring.dispatcher;

import it.uni.bo.monitoring.protocol.server.DataObject;
import it.uni.bo.monitoring.protocol.server.DecomposingPacket;

import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;



public class RawDataProcessing extends  Thread{

	private BlockingQueue<DataObject> rawQueue;
	private ArrayList<BlockingQueue<DataObject>> arrayQueue;
	private DecomposingPacket dp;
	private BlockingQueue<byte[]> rdoToRemove;
	private Logger l;
	
	public RawDataProcessing (BlockingQueue<DataObject> rawQueue, BlockingQueue<byte[]> rdoToRemove, DecomposingPacket dp){
		l = Logger.getLogger(this.getClass());
		arrayQueue=new ArrayList<BlockingQueue<DataObject>>();
		this.rawQueue=rawQueue;
		this.rdoToRemove= rdoToRemove;
		this.dp=dp;
	}
	
	public synchronized void run() {
		while(true){
			try {
					DataObject dataObj=rawQueue.take();
					rawQueue.remove(dataObj);
					if(dataObj!=null){
						dp.parse(dataObj);
						if(dataObj!=null){
							for(int i=0;i<arrayQueue.size();i++)
								arrayQueue.get(i).add(dataObj);
						}
						else{
							rdoToRemove.add(dataObj.getKey());
						}
					}
				
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
	public void addQueue(BlockingQueue<DataObject> queue){
		arrayQueue.add(queue);
	}

}
